package avicit.bdp.dds.server.worker.task;

import avicit.bdp.common.utils.SpringApplicationContext;
import avicit.bdp.dds.dao.entity.TaskInstance;
import avicit.bdp.dds.server.entity.TaskExecutionContext;
import avicit.bdp.dds.server.entity.YarnSubmitConditions;
import avicit.bdp.dds.server.worker.adapter.yarn.YarnAdapter;
import avicit.bdp.dds.service.process.ProcessService;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;

/**
 * @author DIGITAL-MAYANJ
 * abstract yarn task
 */
public abstract class AbstractYarnClientTask extends AbstractTask {

    /**
     * process database access
     */
    protected ProcessService processService;

    protected YarnAdapter yarnAdapter;

    protected String yarnAddress;

    /**
     * Abstract Yarn Task
     *
     * @param taskExecutionContext taskExecutionContext
     * @param logger               logger
     */
    public AbstractYarnClientTask(TaskExecutionContext taskExecutionContext, Logger logger) {
        super(taskExecutionContext, logger);
        this.processService = SpringApplicationContext.getBean(ProcessService.class);
    }


    @Override
    public void handle() throws Exception {
        try {
            YarnSubmitConditions yarnSubmitConditions = buildYarnSubmitConditions();
            appIds = yarnAdapter.submitTask(yarnSubmitConditions);
            if (StringUtils.isNotBlank(appIds)) {
                setExitStatusCode(200);
                setAppIds(appIds);
            }
        } catch (Exception e) {
            logger.error("yarn process failure", e);
            exitStatusCode = -1;
            throw e;
        }
    }

    /**
     * cancel application
     *
     * @param status status
     * @throws Exception exception
     */
    @Override
    public void cancelApplication(boolean status) throws Exception {
        cancel = true;
        TaskInstance taskInstance = processService.findTaskInstanceById(taskExecutionContext.getTaskInstanceId());
        if (status && taskInstance != null) {
            yarnAdapter.killJob(yarnAddress, appIds);
        }
    }

    /**
     * create YarnSubmitConditions
     *
     * @return String
     * @throws Exception exception
     */
    protected abstract YarnSubmitConditions buildYarnSubmitConditions() throws Exception;

    /**
     * set main jar name
     */
    protected abstract void setMainJarName();
}
